package com.lz.zookeeper;

import java.lang.management.ManagementFactory;
import java.lang.reflect.Method;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import com.lz.common.CommonEnum;
import com.lz.common.log4j.Log4jConfig;
import com.lz.common.util.GetCfg;
import com.lz.common.util.StringUtils;

public class ZooKeeperClient {
	
	private static final Logger LOG = LoggerFactory.getLogger(ZooKeeperClient.class);
	private static final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
	private static final String LOG4J_PREFIX = "log4j.";
	private static final Log4jConfig LOG4J_CONFIG = Log4jConfig.getInstance();
	private static final String PORT = GetCfg.getString(CommonEnum.PORT);
	private static final Gson GSON = new Gson();
	private static ZooKeeperClient zooKeeperClient = null;
	private CuratorFramework curator = null;
	/**zk服务注册节点路径**/
	private String zkServerPath;
	/**zk配置节点路径 应用私有配置**/
	private String zkConfigPathPrivate;
	/**zk配置节点路径 公共配置**/
	private String zkConfigPathPublic = "/config/lzmh_public_config";
	/**常量类**/
	private Class<?> constantsClass;
	/**路径节点监听器**/
	private static final Map<String, PathChildrenCache> PATH_LISTENE_MAP = new HashMap<String, PathChildrenCache>(); 
	/**数据节点监听器**/
	private static final Map<String, NodeCache> NODE_LISTENE_MAP = new HashMap<String, NodeCache>(); 
	
	
	private ZooKeeperClient(String zkAddress, String zkServerPath, String zkConfigPathPrivate, Class<?> constantsClass){
		this.zkServerPath = zkServerPath;
		this.zkConfigPathPrivate = zkConfigPathPrivate;
		if(constantsClass != null){
			this.constantsClass = constantsClass;
		}
		if(StringUtils.isNotEmpty(zkAddress)){
			curator = ZooKeeperFactory.get(zkAddress);
		}else{
			curator = ZooKeeperFactory.get();
		}
		
	}

	/**
	 * 获取ZooKeeperClient的实例
	 * @param zkServerPath 注册节点的路径
	 * @param zkConfigPath 配置节点的路径
	 * @return
	 */
	public static synchronized ZooKeeperClient getInstance(String zkServerPath, String zkConfigPath){
		if (zooKeeperClient == null) {    
			zooKeeperClient = new ZooKeeperClient("", zkServerPath, zkConfigPath, null);  
        }    
       return zooKeeperClient; 
	}
	
	/**
	 * 获取ZooKeeperClient的实例
	 * @param zkServerPath 注册节点的路径
	 * @param zkConfigPath 配置节点的路径
	 * @param constantsClass 配置类的Class，必须有init方法，当配置有变化时进行调用进行配置的重新加载
	 * @return
	 */
	public static synchronized ZooKeeperClient getInstance(String zkServerPath, String zkConfigPath, Class<?> constantsClass){
		if (zooKeeperClient == null) {    
			zooKeeperClient = new ZooKeeperClient("", zkServerPath, zkConfigPath, constantsClass);  
		}    
		return zooKeeperClient; 
	}
	/**
	 * 获取ZooKeeperClient的实例
	 * @param zkAddress 	zkServer的Address
	 * @param zkServerPath 注册节点的路径
	 * @param zkConfigPath 配置节点的路径
	 * @param constantsClass 配置类的Class，必须有init方法，当配置有变化时进行调用进行配置的重新加载
	 * @return
	 */
	public static synchronized ZooKeeperClient getInstance(String zkAddress, String zkServerPath, String zkConfigPath, Class<?> constantsClass){
		if (zooKeeperClient == null) {    
			zooKeeperClient = new ZooKeeperClient(zkAddress, zkServerPath, zkConfigPath, constantsClass);  
		}    
		return zooKeeperClient; 
	}
	
	/**
	 * 启动
	 * @throws Exception
	 */
	public void startup(String regValue) throws Exception{
		String name = ManagementFactory.getRuntimeMXBean().getName();  
		String pid = name.split("@")[0];  
		String ipPort = IPUtil.getLocalIP() + ":" + PORT;
		ZookeeperRegContent regContent = new ZookeeperRegContent(ipPort, pid, sdf.format(new Date()));
		regContent.setRegValue(regValue);
		LOG.info("regContent={}", regContent);
		
		//服务注册
		String createServerPath = this.createNode(zkServerPath, GSON.toJson(regContent).getBytes(), CreateMode.EPHEMERAL_SEQUENTIAL);
		LOG.info("createServerPath={}", createServerPath);
		ZooKeeperConnectionStateListener stateListener = new ZooKeeperConnectionStateListener(zkServerPath, regContent);
		curator.getConnectionStateListenable().addListener(stateListener);
		
		//初始化配置(从ZkServer中获取)，私有配置会覆盖统一配置
		initConfigFromZk(zkConfigPathPublic, zkConfigPathPrivate);
		
		//配置节点监听器
		//1.私有配置
		final NodeCache privateConfigNodeCache = new NodeCache(curator, zkConfigPathPrivate);
		privateConfigNodeCache.start();
		NODE_LISTENE_MAP.put(zkConfigPathPrivate, privateConfigNodeCache);
		privateConfigNodeCache.getListenable().addListener(new NodeCacheListener() {
			public void nodeChanged() throws Exception {
				LOG.info("privateConfigNodeCache nodeChanged");
				initConfigFromZk(zkConfigPathPublic, zkConfigPathPrivate);
			}
		});
		//2.公共配置
		final NodeCache publicConfigNodeCache = new NodeCache(curator, zkConfigPathPublic);
		publicConfigNodeCache.start();
		NODE_LISTENE_MAP.put(zkConfigPathPublic, publicConfigNodeCache);
		publicConfigNodeCache.getListenable().addListener(new NodeCacheListener() {
			public void nodeChanged() throws Exception {
				LOG.info("publicConfigNodeCache nodeChanged");
				initConfigFromZk(zkConfigPathPublic, zkConfigPathPrivate);
			}
		});
	}
	
	/**
	 * 初始化配置(从Zk Server中获取)
	 * @param paths
	 * @return
	 */
	private void initConfigFromZk(String ...paths){
		Map<String, String> configMap = new HashMap<String, String>();
		try {
			//从多个配置节点中获取配置
			for(String path : paths){
				String configStr = this.getData(path);
				LOG.info("initConfigFromZk path={} configStr={}", path, configStr);
				if(StringUtils.isNotBlank(configStr)){
					Map<String, String> configMapTemp = GSON.fromJson(configStr, new TypeToken<Map<String, String>>() {}.getType());
					configMap.putAll(configMapTemp);
				}
			}
			LOG.info("initConfigFromZk configMap={}", configMap);
			GetCfg.setConfigMap(configMap);
			//重新初始化配置
			reInitConfig();
			//重新初始化log4j配置(从Zk Server中获取)
			if (LOG4J_CONFIG != null) {
				LOG4J_CONFIG.reSetConfig(configMap, LOG4J_PREFIX);
			}
		} catch (Exception e) {
			e.printStackTrace();
			LOG.error("initConfigFromZk error={}", ExceptionUtils.getFullStackTrace(e));
		}
	}
	
	/**
	 * 通过反射，重新初始化配置
	 * @throws Exception 
	 */
	private void reInitConfig() throws Exception{
		LOG.info("reInitConfig Start");
		if(constantsClass != null){
			String className = constantsClass.getName();
			LOG.info("className={}", className);
			Class<?> cls = Class.forName(className);  
			Method staticMethod = cls.getDeclaredMethod("init"); 
			staticMethod.invoke(cls);
		}
		LOG.info("reInitConfig end");
	}
	
	/**
	 * 新增路径节点监听器,同一个节点调用只执行一次
	 * @param path
	 * @param listener
	 * @throws Exception
	 */
	public void addListener(String path, PathChildrenCacheListener listener) throws Exception{
		if(PATH_LISTENE_MAP.get(path) == null){
			PathChildrenCache cache = new PathChildrenCache(curator, path, true);
			cache.start();
			cache.getListenable().addListener(listener);
			PATH_LISTENE_MAP.put(path, cache);
		}
	}
	/**
	 * 新增数据节点监听器,同一个节点调用只执行一次
	 * @param path
	 * @param listener
	 * @throws Exception
	 */
	public void addListener(String path, NodeCacheListener listener) throws Exception{
		if(NODE_LISTENE_MAP.get(path) == null){
			NodeCache cache = new NodeCache(curator, path);
			cache.start();
			cache.getListenable().addListener(listener);
			NODE_LISTENE_MAP.put(path, cache);
		}
	}
	
	/**
	 * 关闭节点资源
	 * @param nodeCache
	 */
	public void closeNodeCache(NodeCache nodeCache){
		LOG.info("close nodeCache");
		CloseableUtils.closeQuietly(nodeCache);
	}
	
	/**
	 * 停止，关闭资源
	 */
	public void shutdown(){
		LOG.info("close zkClient");
		CloseableUtils.closeQuietly(curator);
		LOG.info("close configNodeCache");
		for (NodeCache value : NODE_LISTENE_MAP.values()) {  
			LOG.info("close configNodeCache=" + value);
			CloseableUtils.closeQuietly(value);
		}  
		for (PathChildrenCache value : PATH_LISTENE_MAP.values()) {  
			LOG.info("close configNodeCache=" + value);
			CloseableUtils.closeQuietly(value);
		}  
	}
	
	/**
	 * 创建节点
	 * @param zkClient
	 * @param path
	 * @param datas
	 * @param mode
	 * @return
	 * @throws Exception 
	 */
	public String createNode(String path, byte[] datas, CreateMode mode) throws Exception{
		return ZooKeeperUtil.createNode(curator, path, datas, mode);
	}
	/**
	 * 设置节点数据，如果节点不存在则创建节点
	 * @param curator
	 * @param path
	 * @param datas
	 * @return
	 * @throws Exception
	 */
	public Stat setData(String path, byte[] datas, CreateMode mode) throws Exception{
		if(ZooKeeperUtil.checkExists(curator, path) == null){
			ZooKeeperUtil.createNode(curator, path, datas, mode);
		}
		return ZooKeeperUtil.setData(curator, path, datas);
	}
	
	/**
	 * 获取子节点
	 * @param zkClient
	 * @param path
	 * @return
	 * @throws Exception
	 */
	public List<String> getChildren(String path) throws Exception{
		return ZooKeeperUtil.getChildren(curator, path);
	}
	
	/**
	 * 获取节点数据
	 * @param zkClient
	 * @param path
	 * @return
	 * @throws Exception
	 */
	public String getData(String path) throws Exception{
		return ZooKeeperUtil.getData(curator, path);
	}	
	
	/**
	 * 删除节点
	 * @param zkClient
	 * @param path
	 * @throws Exception
	 */
	public void deleteNode(String path) throws Exception{
		ZooKeeperUtil.deleteNode(curator, path);
	}
	
	/**
	 * 检查节点是否存在
	 * @param zkClient
	 * @param path
	 * @return
	 * @throws Exception
	 */
	public Stat checkExists(String path) throws Exception{
		return ZooKeeperUtil.checkExists(curator, path);
	}	
	
}
